feat(cdk): Add RecordExpander component for nested array extraction

Summary

This PR adds a new RecordExpander component to the CDK that enables extracting items from nested array fields and emitting each item as a separate record. This is needed to fix the Stripe invoice_line_items and subscription_items stream issues where the events endpoint returns parent objects with nested arrays, but we need to emit each child item as a separate record.

Key changes:

• New RecordExpander class in airbyte_cdk/sources/declarative/expanders/
• New OnNoRecords enum for type-safe on_no_records behavior
• Integration with DpathExtractor via optional record_expander parameter
• Support for wildcard paths (e.g., ["sections", "*", "items"])
• Optional remain_original_record flag to embed parent record context
• on_no_records parameter: OnNoRecords.skip (default) or OnNoRecords.emit_parent behavior
• Schema updates and auto-generated models
• 19 new unit tests covering all expansion scenarios

Example usage:

extractor:
  type: DpathExtractor
  field_path: ["data", "object"]
  record_expander:
    type: RecordExpander
    expand_records_from_field: ["items", "data"]
    on_no_records: skip

To copy specific parent fields into expanded child records, use existing RecordTransformations (e.g., AddFields) downstream of the extractor rather than configuring it on RecordExpander directly.

Updates since last revision

• Removed ParentFieldMapping and parent_fields_to_copy: Parent field copying is now delegated to existing RecordTransformations (e.g., AddFields), keeping RecordExpander focused on expansion only
• Added OnNoRecords enum: on_no_records is now a proper enum (OnNoRecords.skip, OnNoRecords.emit_parent) instead of a raw string, for type safety
• Lint and format auto-fixes applied

Review & Testing Checklist for Human

This is a YELLOW risk PR (medium confidence). Please verify:

• on_no_records behavior: Verify the emit_parent option correctly emits the parent record when expansion path is missing, empty, or non-array. The logic at the end of expand_record() in record_expander.py handles this.

• End-to-end testing: This PR only includes unit tests. The real-world behavior needs to be verified with the Stripe connector in the companion PR (fix(source-stripe): Fix invoice_line_items and subscription_items incremental streams (do not merge) airbyte#70294). In particular, confirm that parent field copying via RecordTransformations works as expected now that ParentFieldMapping has been removed.

Recommended test plan:

1. Run CDK tests: poetry run pytest unit_tests/sources/declarative/extractors/test_dpath_extractor.py -v
2. Test with Stripe connector (separate PR) to verify end-to-end behavior for both subscription_items and invoice_line_items streams
3. Try edge cases: empty arrays, missing paths, remain_original_record with on_no_records: emit_parent

Notes

• Related issues:
  • https://github.com/airbytehq/oncall/issues/8683 (invoice_line_items)
  • https://github.com/airbytehq/oncall/issues/10756 (subscription_items)
• Stripe connector PR: fix(source-stripe): Fix invoice_line_items and subscription_items incremental streams (do not merge) airbyte#70294 (blocked on this PR)
• All 19 new unit tests passing locally
• MyPy type checking passes
• Lint checks pass

Design

Design Document: RecordExpander Component

PR: #859
Related issues: oncall#8683 (invoice_line_items), oncall#10756 (subscription_items)
Companion PR: airbytehq/airbyte#70294 (Stripe connector)

1. Problem Statement

Several Stripe streams (invoice_line_items, subscription_items) use an events-based incremental sync that fetches parent objects (e.g., an Invoice) from the events endpoint. These parent objects contain nested arrays of child items (e.g., invoice.lines.data[]). The connector needs to emit each child item as a separate record, not the parent object.

Before this PR, the DpathExtractor could only extract records at a single path depth. It had no mechanism to "explode" a nested array within an extracted record into multiple output records. Connector developers had to write custom Python code to handle this pattern, which is common across many APIs.

2. Solution Overview

The PR introduces a new RecordExpander declarative component that plugs into the existing DpathExtractor. When configured, it takes each record extracted by DpathExtractor and expands it by pulling out items from a nested array field, emitting each item as a separate record.

Parent field copying (e.g., copying invoice_id from the parent into each child) is intentionally not part of RecordExpander. Instead, use existing RecordTransformations such as AddFields downstream of the extractor to enrich expanded records with parent context.

Data flow with RecordExpander:

API Response
    |
    v
DpathExtractor.extract_records()
    |  Extracts top-level records via field_path (e.g., "data.object")
    v
[Parent Record 1, Parent Record 2, ...]
    |
    v  (for each parent record)
RecordExpander.expand_record()
    |  Extracts child items from nested array (e.g., "items.data")
    v
[Child Item 1a, Child Item 1b, ..., Child Item 2a, ...]

Data flow without RecordExpander (existing behavior, unchanged):

API Response
    |
    v
DpathExtractor.extract_records()
    |  Extracts records via field_path
    v
[Record 1, Record 2, ...]  (emitted directly)

3. Component Architecture

3.1 New Classes

OnNoRecords (Enum)

Location: airbyte_cdk/sources/declarative/expanders/record_expander.py

Defines the behavior when record expansion produces no records:

• skip — Emits nothing (default)
• emit_parent — Emits the original parent record unchanged

RecordExpander (dataclass)

Location: airbyte_cdk/sources/declarative/expanders/record_expander.py

The core component. Given a parent record, it navigates to a nested array field and yields each element as a separate record.

Attributes:

Key method - expand_record(record):

expand_record(record) -> Iterable[MutableMapping]

1. Evaluates the expand_records_from_field path (resolving interpolation).
2. If the path contains a wildcard (*), uses dpath.values() to match multiple nested arrays.
3. Otherwise, uses dpath.get() to retrieve the single nested value.
4. For each item in the extracted array:
   • If the item is a dict, creates a shallow copy and applies parent context (via _apply_parent_context).
   • Otherwise, yields the item as-is.
5. If no items were expanded and on_no_records == OnNoRecords.emit_parent, yields the original parent record.

3.2 Modified Classes

DpathExtractor

A new optional attribute record_expander: Optional[RecordExpander] = None is added. The extract_records() method is modified:

• Without record_expander: Behavior is identical to before (no change).
• With record_expander: After extracting each record from the response body, each record is passed through record_expander.expand_record(), and all expanded child records are yielded instead.

# Simplified logic in extract_records():
if isinstance(extracted, list):
    if not self.record_expander:
        yield from extracted                          # original behavior
    else:
        for record in extracted:
            yield from self.record_expander.expand_record(record)  # new behavior

ModelToComponentFactory

• create_dpath_extractor(): Now checks for model.record_expander and instantiates a RecordExpander if present.
• New create_record_expander() method: Converts the Pydantic model to a RecordExpander instance, mapping on_no_records string values to the OnNoRecords enum.

declarative_component_schema.yaml

New schema definitions added:

• RecordExpander: Defines expand_records_from_field, remain_original_record, and on_no_records (enum: skip/emit_parent).
• DpathExtractor updated: New optional record_expander property referencing RecordExpander.

Auto-generated Pydantic models

declarative_component_schema.py updated with generated RecordExpander and OnNoRecords model classes.

manifest_component_transformer.py

Updated to include propagation mapping for DpathExtractor.record_expander, ensuring parameters are correctly propagated during manifest resolution.

4. YAML Manifest Usage

Basic expansion

extractor:
  type: DpathExtractor
  field_path: ["data", "object"]
  record_expander:
    type: RecordExpander
    expand_records_from_field: ["lines", "data"]

This extracts response.data.object as the parent record, then expands parent.lines.data[] into individual records.

With wildcard path and parent embedding

record_expander:
  type: RecordExpander
  expand_records_from_field: ["sections", "*", "items"]
  remain_original_record: true
  on_no_records: emit_parent

Matches sections[0].items[], sections[1].items[], etc. Each child gets an "original_record" key containing the full parent. If no items are found, the parent is emitted unchanged.

5. Concrete Example: Stripe invoice_line_items

API response from Stripe events endpoint:

{
  "data": {
    "object": {
      "id": "in_123",
      "customer": "cus_456",
      "lines": {
        "data": [
          {"id": "il_aaa", "amount": 1000, "description": "Widget"},
          {"id": "il_bbb", "amount": 2000, "description": "Gadget"}
        ]
      }
    }
  }
}

Manifest configuration:

extractor:
  type: DpathExtractor
  field_path: ["data", "object"]
  record_expander:
    type: RecordExpander
    expand_records_from_field: ["lines", "data"]

Output records:

{"id": "il_aaa", "amount": 1000, "description": "Widget"}
{"id": "il_bbb", "amount": 2000, "description": "Gadget"}

To enrich these with the parent invoice_id, use a downstream AddFields transformation rather than configuring it on RecordExpander.

6. Edge Cases and Behavior Matrix

7. Design Decisions and Trade-offs

Why a separate RecordExpander class instead of inline logic in DpathExtractor?

Separation of concerns. DpathExtractor handles response-level extraction (navigating JSON to find records). RecordExpander handles record-level transformation (flattening nested arrays). This keeps each class focused and testable independently, and allows RecordExpander to potentially be reused with other extractor types in the future.

Why dpath for nested access?

The CDK already uses dpath extensively for path-based JSON navigation. Reusing it maintains consistency and avoids introducing new dependencies. The wildcard (*) support comes for free from dpath.values().

Why on_no_records instead of always emitting parent?

Different use cases need different behavior. For Stripe invoice_line_items, if an invoice event has no line items, we want to skip it entirely (skip). For other APIs, the parent record itself might be the meaningful output when no children exist (emit_parent).

Why no parent_fields_to_copy on RecordExpander?

Parent field copying was originally part of this component but was removed to keep RecordExpander focused solely on expansion. Copying parent fields into child records is already supported by existing RecordTransformations (e.g., AddFields), which is the idiomatic CDK pattern for record enrichment. This avoids duplicating transformation logic and keeps the component composable.

Why an OnNoRecords enum instead of raw strings?

Using an enum provides type safety, IDE autocompletion, and prevents invalid values at construction time rather than at runtime.

Shallow copy behavior

dict(item) creates a shallow copy when yielding expanded child records. This means nested mutable objects are shared between the parent and child records. This is acceptable because:

1. Records are generally treated as read-only after extraction.
2. A deep copy would be expensive for large records.

8. Files Changed

9. Testing

19 unit tests covering:

• Basic expansion from nested arrays
• Wildcard path matching
• remain_original_record flag
• on_no_records with both OnNoRecords.skip and OnNoRecords.emit_parent
• Missing/empty/non-array nested paths
• Non-dict items in nested arrays
• Multiple parent records
• Integration with DpathExtractor.extract_records()

End-to-end validation with the Stripe connector is covered by the companion PR (airbyte/airbyte#70294).

Link to Devin session: https://app.devin.ai/sessions/08169cc37f9342acb410071ab8306f05
Requested by: Alfredo Garcia (@agarctfi)